package io.kiki.stack.netty.processor;


import io.kiki.stack.netty.channelManager.Channel;
import io.kiki.stack.netty.channelManager.ChannelContext;
import io.kiki.stack.netty.command.Command;
import io.kiki.stack.netty.command.HeartbeatRequestCommand;
import io.kiki.stack.netty.command.HeartbeatResponseCommand;
import io.kiki.stack.netty.invoker.Future;
import io.kiki.stack.netty.util.InetSocketAddressUtil;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;


@Slf4j
@Setter
@Getter
public class HeartBeatProcessor extends AbstractProcessor {

    @Override
    public void doProcess(final ChannelContext channelContext, Command command) {
        if (command instanceof HeartbeatRequestCommand) {// process the heartbeat
            final int id = command.getId();
            HeartbeatResponseCommand heartbeatResponseCommand = new HeartbeatResponseCommand();
            heartbeatResponseCommand.setId(id);
            channelContext.writeAndFlush(heartbeatResponseCommand).addListener(new ChannelFutureListener() {

                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        log.error("Send heartbeat ack failed! Id={}, to remoteAddr={}", id, InetSocketAddressUtil.parseRemoteAddress(channelContext.getChannelHandlerContext().channel()));
                    }
                }
            });
        } else
            if (command instanceof HeartbeatResponseCommand) {
                Channel channel = channelContext.getChannelHandlerContext().channel().attr(Channel.CHANNEL).get();
                Future future = channel.removeFuture(command.getId());
                if (future != null) {
                    future.putResponse(command);
                    future.cancelTimeout();
                    try {
                        future.executeInvokeCallback();
                    } catch (Exception e) {
                        log.error("Exception caught when executing heartbeat invoke callback. From {}", InetSocketAddressUtil.parseRemoteAddress(channelContext.getChannelHandlerContext().channel()), e);
                    }
                } else {
                    log.warn("Cannot find heartbeat InvokeFuture, maybe already timeout. Id={}, From {}", command.getId(), InetSocketAddressUtil.parseRemoteAddress(channelContext.getChannelHandlerContext().channel()));
                }
            } else {
                throw new RuntimeException("Cannot process command: " + command.getClass().getName());
            }
    }

}